package mergeFile;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.IOException;
import java.net.URI;
import java.util.Date;


class myPathFilter implements PathFilter {  //过滤掉文件名满足特定条件的文件
    String reg = null;

    myPathFilter(String reg) {
        this.reg = reg;
    }

    public boolean accept(Path path) {
        return !(path.toString().matches(reg));
    }

}

public class merge {
    Path inputPath = null;  //待合并的文件所在的目录的路径
    Path outputPath = null; //输出文件的路径

    public merge(String input, String output) {
        this.inputPath = new Path(input);
        this.outputPath = new Path(output);
    }

    public void doMerge() throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://master:8020");
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");

        FileSystem fsSource = FileSystem.get(URI.create(inputPath.toString()), conf);
        FileSystem fsDst = FileSystem.get(URI.create(outputPath.toString()), conf);

        FileStatus[] sourceStatus = fsSource.listStatus(inputPath, new myPathFilter("\\..*"));
        FSDataOutputStream fsdos = fsDst.create(outputPath);

        //下面分别读取过滤之后的每个文件的内容，并输出到同一个文件中
        for (FileStatus sta : sourceStatus) {
//            System.out.println("路径: " + sta.getPath() + "   文件大小: " + sta.getLen() + "   权限: " + sta.getPermission() + "  内容: ");
            FSDataInputStream fsdis = fsSource.open(sta.getPath());
            byte[] data = new byte[1024];
            int read = -1;
//            PrintStream ps = new PrintStream(System.out);
            while ((read = fsdis.read(data)) > 0) {
//                ps.write(data, 0, read);
                fsdos.write(data, 0, read);
            }
        }
        fsdos.close();
    }

    public static void main(String[] args) throws IOException {
        Date date = new Date();
        long time = date.getTime();
        merge merge = new merge("hdfs://master:8020/tmp/project/test/result", "hdfs://master:8020/tmp/project/test/result.txt");
        merge.doMerge();
        System.out.println(date.getTime() - time);
    }
}